import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object Job3 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName(this.getClass.getCanonicalName)
      .master("local[*]")
      .getOrCreate()

    val sc = spark.sparkContext

    val clickRDD = sc.textFile("D:\\MyStudent\\studentCode\\SparkDemo\\JobDemo\\data\\click.log")

    val impRDD = sc.textFile("D:\\MyStudent\\studentCode\\SparkDemo\\JobDemo\\data\\imp.log")

    val clickRes = clickRDD.map{line => {
      val arr = line.split("\\s+")
      val adid = arr(3).substring(arr(3).lastIndexOf("=") + 1)
      (adid, 1)
    }}.reduceByKey(_ + _)

    val impRes = impRDD.map { line =>
      val arr = line.split("\\s+")
      val adid = arr(3).substring(arr(3).lastIndexOf("=") + 1)
      (adid, 1)
    }.reduceByKey(_ + _)


    clickRes.union(impRes)
      .map((x) => (x._1,x._2.toString))
      .reduceByKey(_ + "," + _)
      .repartition(1)
      .saveAsTextFile("D:\\MyStudent\\studentCode\\SparkDemo\\JobDemo\\data\\job3\\add_log")

    sc.stop()

  }
}
